package com.distributer.timer.init;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import com.distributer.timer.zookeeper.ZookeeperClient;
import com.distributer.timer.zookeeper.ZookeeperClientListener;
import com.distributer.timer.zookeeper.ZookeeperConfig;

/**
 * 主节点注册
 * @author yilin
 *
 */
@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class ZkApplicationRunner implements ApplicationRunner {

    private final static Logger log = LoggerFactory.getLogger(ZkApplicationRunner.class);

    @Autowired
    private ZookeeperClientListener zkClientListener;

    @Autowired
    private ZookeeperConfig zookeeperConfig;

    @Override
    public void run(ApplicationArguments applicationArguments) throws Exception {
        log.error("启动zookeeper: addr:{}, sleepTime:{}, max:{}, connectionTime:{}", zookeeperConfig.getAddr(), zookeeperConfig.getSleepTime(), zookeeperConfig.getMax(), zookeeperConfig.getConnectionTime());
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(zookeeperConfig.getAddr())
                .retryPolicy(new ExponentialBackoffRetry(zookeeperConfig.getSleepTime(), zookeeperConfig.getMax()))
                .connectionTimeoutMs(zookeeperConfig.getConnectionTime()).build();
        LeaderLatch leaderLatch = new LeaderLatch(client, zookeeperConfig.getRootPath(), zookeeperConfig.getParticipantId(), LeaderLatch.CloseMode.NOTIFY_LEADER);
        leaderLatch.addListener(zkClientListener);
        ZookeeperClient zkClient = new ZookeeperClient(leaderLatch, client);
        try {
            zkClient.startZKClient();
        } catch (Exception e) {
            log.error("======>>>>>>zk客户端连接失败<<<<<=====error:{}===", e);
            return;
        }
        CuratorFrameworkState state = client.getState();
        if (CuratorFrameworkState.STOPPED == state) {
            log.error("zk客户端已关闭");
            return;
        }
    }
}
